package light.network;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

class TcpEvent {
	static final int Accept = 0;
	static final int Receive = 1;
	static final int Exception = 2;
	int type;
	AsynchronousSocketChannel socket;
	Object attachment;
	ByteBuffer buffer;
}

enum TcpEpType {
	Server, Client
}

public class TcpEndpoint implements Endpoint {

	int bufferSize = 1 * 1024;
	AsynchronousServerSocketChannel server;
	AsynchronousSocketChannel cp;
	TcpEpType epType = TcpEpType.Client;
	IEndpointHandler handler;

	Deque<TcpEvent> completes = new ArrayDeque<TcpEvent>(1024);
	Map<String, AsynchronousSocketChannel> currentConnection = new HashMap<String, AsynchronousSocketChannel>();

	boolean processQueue = true;

	public TcpEndpoint() throws IOException {
	}

	@Override
	public void setReceiveHandler(IEndpointHandler handler) {
		this.handler = handler;
	}

	public void sendTcp(Packet pack, SocketAddress target) throws IOException {
		AsynchronousSocketChannel c = currentConnection.get(target);
		if (c == null)
			throw new IOException("connection is not established.");
		else {
			c.write(pack.buffer());
		}
	}

	public void sendTcp(Packet pack, SocketAddress target, Object attachment, CompletionHandler<Integer, ? super Object> h) throws IOException {
		AsynchronousSocketChannel c = currentConnection.get(target);
		if (c == null)
			throw new IOException("connection is not established.");
		else {
			c.write(pack.buffer(), attachment, h);
		}
	}

	private void start() {
		epType = TcpEpType.Server;
		server.accept(null, new AcceptHandler());
		// 处理线程 keep-alive
		new Thread() {
			@Override
			public void run() {
				while (processQueue) {
					TcpEvent s = completes.pollFirst();
					if (s == null) {
						try {
							Thread.sleep(100);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					} else {
						switch (s.type) {
						case TcpEvent.Receive:
							handler.onReceive(new TcpEpSession(s.socket), TcpPacket.wrap(s.buffer));
							break;
						case TcpEvent.Accept:
							handler.onAccept(new TcpEpSession(s.socket));
							break;
						case TcpEvent.Exception:
							handler.onFail(new TcpEpSession(s.socket), null);
							break;
						default:
							break;
						}
					}
					Thread.yield();
				}
			};
		}.start();
	}

	@Override
	public void start(SocketAddress address) throws IOException {
		epType = TcpEpType.Server;
		server = AsynchronousServerSocketChannel.open();
		server.bind(address);
		start();
	}

	@Override
	public void bind(SocketAddress address) throws IOException {
		epType = TcpEpType.Client;
		cp = AsynchronousSocketChannel.open();
		cp.bind(address);
	}

	@Override
	public void stop() {
		processQueue = false;
	}

	class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
		@Override
		public void completed(AsynchronousSocketChannel result, Object attachment) {
			server.accept(null, new AcceptHandler());
			ByteBuffer dst = ByteBuffer.allocate(bufferSize);
			result.read(dst, null, new ReceiveCompHandler().ep(result).bf(dst));
			TcpEvent e = new TcpEvent();
			e.socket = result;
			e.attachment = null;
			e.type = TcpEvent.Accept;
			completes.addLast(e);
		}

		@Override
		public void failed(Throwable exc, Object attachment) {
			exc.printStackTrace();
		}
	}

	class ReceiveCompHandler implements CompletionHandler<Integer, Object> {
		AsynchronousSocketChannel client;
		ByteBuffer dst;
		ByteBuffer l2Buffer = ByteBuffer.allocate(bufferSize);
		int lackLen = 0;

		ReceiveCompHandler ep(AsynchronousSocketChannel ep) {
			client = ep;
			return this;
		}

		ReceiveCompHandler bf(ByteBuffer dst) {
			this.dst = dst;
			return this;
		}

		void process(ByteBuffer dst) {
			int recvLen = dst.limit() - dst.position();
			if (lackLen <= 0) {
				int len = TcpPacket.validateLength(dst);
				if(len==-1)
					return;
				if (len == recvLen) {
					TcpEvent e = new TcpEvent();
					e.type = TcpEvent.Receive;
					e.socket = client;
					e.buffer = dst;
					completes.addLast(e);
					dst = ByteBuffer.allocate(bufferSize);
				} else if (len < recvLen) {
					TcpEvent e = new TcpEvent();
					e.type = TcpEvent.Receive;
					e.socket = client;
					e.buffer = ByteBuffer.wrap(dst.array(), dst.position(), len);// dst;
					completes.addLast(e);
					dst.position(len);
					process(dst);
				} else// if(len>recvLen)
				{
					l2Buffer = ByteBuffer.allocate(bufferSize);
					dst.position(0);
					l2Buffer.put(dst);
					lackLen = (len - recvLen);
				}
			} else {
				if (recvLen == lackLen) {
					l2Buffer.put(dst);
					l2Buffer.flip();
					TcpEvent e = new TcpEvent();
					e.type = TcpEvent.Receive;
					e.socket = client;
					e.buffer = l2Buffer;
					completes.addLast(e);
					lackLen = 0;
				} else if (recvLen < lackLen) {
					l2Buffer.put(dst);
					lackLen -= recvLen;
				} else// if(recvLen>lackLen)
				{
					l2Buffer.put(dst.array(), dst.position(), dst.position() + lackLen);
					l2Buffer.flip();
					TcpEvent e = new TcpEvent();
					e.type = TcpEvent.Receive;
					e.socket = client;
					e.buffer = l2Buffer;
					l2Buffer = null;
					completes.addLast(e);
					dst.position(dst.position() + lackLen);
					lackLen = 0;
					process(dst);
				}
			}
		}

		@Override
		public void completed(Integer result, Object a) {
			ByteBuffer buf = ByteBuffer.allocate(bufferSize);
			dst.flip();
			process(dst);
			client.read(buf, null, this.bf(buf));// new
													// ReceiveCompHandler().setClientEp(client).setBuffer(buf)
		}

		@Override
		public void failed(Throwable exc, Object a) {
			exc.printStackTrace();
		}
	}

	@Override
	public int getBufferSize() {
		return bufferSize;
	}

	@Override
	public void setBufferSize(int bufferSize) {
		this.bufferSize = bufferSize;
	}

	@Override
	public void close() throws IOException {
		server.close();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see light.network.Endpoint#createSession(java.net.SocketAddress)
	 */
	@Override
	public void openSession(SocketAddress target, Object attachment, final SendHandler h) {
		if (epType == TcpEpType.Client) {
			cp.connect(target, attachment, new CompletionHandler<Void, Object>() {
				@Override
				public void completed(Void result, Object attachment) {
					h.onSession(new TcpEpSession(cp), attachment);
				}

				@Override
				public void failed(Throwable exc, Object attachment) {
					h.onFailed(exc, attachment);
				}
			});
		} else {
			AsynchronousSocketChannel cl = currentConnection.get(target);
			if (cl != null)
				h.onSession(new TcpEpSession(cl), attachment);
			else
				h.onFailed(new IOException("client connection is not established."), attachment);
		}
	}

	public static void main(String[] args) throws Exception {

		TcpEndpoint s = new TcpEndpoint();
		s.start(new InetSocketAddress(10606));
		s.setReceiveHandler(new EndpointHandlerAdapter() {
			@Override
			public void onReceive(EndpointSession ep, Packet packet) {
				try {
					System.out.println("recv: " + packet);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
		runClient(1);
	}

	@Override
	public SocketAddress getLocalAddress() throws IOException {
		return server.getLocalAddress();
	}

	static void runClient(int num) throws Exception {
		for (int i = 0; i < 10; i++) {
			Endpoint s = new TcpEndpoint();
			s.setReceiveHandler(new EndpointHandlerAdapter() {
				@Override
				public void onReceive(EndpointSession ep, Packet packet) {
					System.err.println("client recv: " + packet.getLength());
				}
			});
			s.bind(new InetSocketAddress("localhost", 10607 + num + i * 1000));
			s.openSession(new InetSocketAddress("localhost", 10606), null, new SendHandler() {
				@Override
				public void onSession(EndpointSession session, Object attachment) {
					try {
						ByteBuffer unit = ByteBuffer.allocate(32).putInt(0).putInt(18).put("packetunit".getBytes());
						ByteBuffer tol = ByteBuffer.allocate(100);
						unit.flip();
						tol.put(unit);
						unit.flip();
						tol.put(unit);
						tol.flip();
//						ByteBuffer buf = (ByteBuffer) (ByteBuffer.allocate(32).putInt(0).putInt(8+8).put("12345".getBytes()).flip());
						session.send(tol);
						Thread.sleep(200);
//						ByteBuffer buf2 = (ByteBuffer) (ByteBuffer.allocate(32).put("678".getBytes()).flip());
						//session.send(buf2);
						Thread.sleep(200);
						//.send(TcpPacket.allocate(16).put("I am two!".getBytes()).flip());
					} catch (Exception e) {
						e.printStackTrace();
					}
				}

				@Override
				public void onFailed(Throwable exc, Object attachment) {
					exc.printStackTrace();
				}
			});

		}
	}

	@Override
	public Packet allocatePacket(int capacity) {
		return TcpPacket.allocate(capacity);
	}

	@Override
	public Packet wrapPacket(ByteBuffer src) {
		return TcpPacket.wrap(src);
	}
}

class TcpEpSession extends EndpointSession {
	AsynchronousSocketChannel ep;

	public TcpEpSession(AsynchronousSocketChannel ep) {
		this.ep = ep;
	}

	@Override
	public void send(Packet packet, Object attachment, final CompletionHandler<Integer, Object> handler) throws IOException {
		ByteBuffer[] data = ((TcpPacket) packet).genPacketBuffer();
		ep.write(data, 0, data.length, 30, TimeUnit.SECONDS, attachment, new CompletionHandler<Long, Object>() {
			@Override
			public void completed(Long result, Object attachment) {
				handler.completed(result.intValue(), attachment);
			}

			@Override
			public void failed(Throwable exc, Object attachment) {
				handler.failed(exc, attachment);
			}
		});
	}

	@Override
	public void close() throws IOException {
		ep.close();
	}

	@Override
	public SocketAddress getRemoteAddress() throws IOException {
		return ep.getRemoteAddress();
	}

	@Override
	public SocketAddress getLocalAddress() throws IOException {
		return ep.getLocalAddress();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see light.network.EndpointSession#send(light.network.Packet)
	 */
	@Override
	public void send(Packet packet) throws IOException {
		ByteBuffer[] data = ((TcpPacket) packet).genPacketBuffer();
		ep.write(data, 0, data.length, 30, TimeUnit.SECONDS, null, new CompletionHandler<Long, Object>() {
			@Override
			public void completed(Long result, Object attachment) {
			}

			@Override
			public void failed(Throwable exc, Object attachment) {
				exc.printStackTrace();
			}
		});
	}

	@Override
	public void send(ByteBuffer buf) throws IOException {
		ep.write(buf);
	}
}